package deng.camel.core;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;

import javax.naming.NamingException;

import org.apache.camel.Processor;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.management.DefaultManagementStrategy;
import org.apache.camel.util.jndi.JndiContext;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import deng.camel.beans.processors.CountDownLatchProcessor;

/**
 * This base class allow subclass to quickly setup a Camel context and try out the 
 * Camel Ride experience!
 * 
 * Subclass just need to provide a RouteBuilder in a {@link #configure()} method 
 * and then invoke {@link #start()} in their main method.
 * 
 * The nature of camel route process will run on separate threads and needs to be
 * shutdown properly after message are processed. We provide a simple shutdownLatch
 * in this base class to control this. By default, the camel context created here 
 * will not shutdown until the shutdownLatch has been countDown at least once. 
 * Subclass needs to do ensure to decrement this latch at some point after their 
 * route processing. Simply adding this at the end of your route will work:
 *   <code> 
 *   from("someEndpoint").
 *   ... // more pipeline route, then at last add this
 *   process(getShutdownLatchProcessor());
 *   </code>
 *    
 * Subclass may also set their own instance of shutdownLatch to whatever
 * number they want using {@link #setShutdownLatch(CountDownLatch)}.
 * 
 * @author dengz1
 *
 */
public abstract class CamelRide extends RouteBuilder {
	
	protected Log logger = LogFactory.getLog(getClass());
	private CountDownLatch shutdownLatch;
	private JndiContext beanRegistry;
	private DefaultCamelContext camelContext;
	private Processor shutdownLatchProcessor;
	
	public void setShutdownLatch(CountDownLatch shutdownLatch) {
		this.shutdownLatch = shutdownLatch;
	}
	
	public Processor getShutdownLatchProcessor() {
		return shutdownLatchProcessor;
	}
	public void setShutdownLatchProcessor(Processor shutdownLatchProcessor) {
		this.shutdownLatchProcessor = shutdownLatchProcessor;
	}
	
	public DefaultCamelContext getCamelContext() {
		return camelContext;
	}
	
	/** Subclass needs to invoke this in their Main method. */
	public void start() {
		try {
			logger.debug("Hello, welcome to the Camel Ride!");
			
			// Initialize default latch value
			if (shutdownLatch == null) {
				logger.trace("Initializing shutdownLatch to have 1 count down.");
				shutdownLatch = new CountDownLatch(1);
			}
			
			// Initialize default shutdownLatchProcessor
			if (shutdownLatchProcessor == null) {
				shutdownLatchProcessor = new CountDownLatchProcessor(shutdownLatch);
			}
			
			// Create and setup CamelContext
			logger.info("Create and setup Camel context instance and route.");
			beanRegistry = new JndiContext();
			camelContext = new DefaultCamelContext(beanRegistry);
			camelContext.setManagementStrategy(new DefaultManagementStrategy()); // (Non JMX - prevent default warning)
			camelContext.addRoutes(this);

			logger.debug("Camel started. Enjoy the ride!");
			
			beforeCamelStart();
			camelContext.start();
			afterCamelStart();
			
			// Wait for shutdown latch.
			logger.trace("Main thread will waiting for shutdownLatch to countDown " + shutdownLatch.getCount() + " time(s).");
			shutdownLatch.await();

			logger.debug("The Camel ride is finished. Shutting down now.");
			camelContext.shutdown();
			
			logger.debug("We are done. Bye bye.");
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}
		
	/** Let subclass register any bean object to be use in their route. */
	public void registerBean(String beanName, Object beanInstance) {
		try {
			beanRegistry.bind(beanName, beanInstance);
		} catch (NamingException e) {
			throw new RuntimeException(e);
		}
	}
	
	/** Let subclass quickly send a message body and/or headers to a endpointUri */
	public void sendMessage(String endpointUri, Object body, Object ... pairArgs) {
		ProducerTemplate producer = camelContext.createProducerTemplate();
		logger.info("Sending message to " + endpointUri);
		producer.sendBodyAndHeaders(endpointUri, body, mkHeaders(pairArgs));
	}
	
	/** Just a helper method to create a HashMap */
	public Map<String, Object> mkHeaders(Object ... objects) {
		HashMap<String, Object> map = new HashMap<String, Object>(objects.length);
		int i = 0;
		while(i < objects.length) {
			map.put((String)objects[i++], objects[i++]);
		}
		return map;
	}

	/** A callback method in case subclass can override to custom work. */
	public void beforeCamelStart() {
	}

	/** A callback method in case subclass can override to custom work. */
	public void afterCamelStart() {
	}
}
